package com.sunday.common.mq.rocket.brave.study.base.e8_Skip;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;

import java.util.concurrent.TimeUnit;

/**
 * https://github.com/apache/rocketmq/blob/master/docs/cn/best_practice.md#3---%E8%B7%B3%E8%BF%87%E9%9D%9E%E9%87%8D%E8%A6%81%E6%B6%88%E6%81%AF
 * <p>
 * 3 跳过非重要消息
 * 发生消息堆积时，如果消费速度一直追不上发送速度，如果业务对数据要求不高的话，可以选择丢弃不重要的消息。例如，当某个队列的消息数堆积到100000条以上，则尝试丢弃部分或全部消息，这样就可以快速追上发送消息的速度。示例代码如下：
 * <p>
 * public ConsumeConcurrentlyStatus consumeMessage(
 * List<MessageExt> msgs,
 * ConsumeConcurrentlyContext context) {
 * long offset = msgs.get(0).getQueueOffset();
 * String maxOffset =
 * msgs.get(0).getProperty(Message.PROPERTY_MAX_OFFSET);
 * long diff = Long.parseLong(maxOffset) - offset;
 * if (diff > 100000) {
 * // TODO 消息堆积情况的特殊处理
 * return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
 * }
 * // TODO 正常消费过程
 * return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
 * }
 *
 * 2.3 消费打印日志
 * 如果消息量较少，建议在消费入口方法打印消息，消费耗时等，方便后续排查问题。
 *
 *    public ConsumeConcurrentlyStatus consumeMessage(
 *             List<MessageExt> msgs,
 *             ConsumeConcurrentlyContext context) {
 *         log.info("RECEIVE_MSG_BEGIN: " + msgs.toString());
 *         // TODO 正常消费过程
 *         return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
 *     }
 * 如果能打印每条消息消费耗时，那么在排查消费慢等线上问题时，会更方便。
 *
 * 2.4 其他消费建议
 * 1 关于消费者和订阅
 * ​第一件需要注意的事情是，不同的消费者组可以独立的消费一些 topic，并且每个消费者组都有自己的消费偏移量，请确保同一组内的每个消费者订阅信息保持一致。
 *
 * 2 关于有序消息
 * 消费者将锁定每个消息队列，以确保他们被逐个消费，虽然这将会导致性能下降，但是当你关心消息顺序的时候会很有用。我们不建议抛出异常，你可以返回 ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT 作为替代。
 *
 * 3 关于并发消费
 * 顾名思义，消费者将并发消费这些消息，建议你使用它来获得良好性能，我们不建议抛出异常，你可以返回 ConsumeConcurrentlyStatus.RECONSUME_LATER 作为替代。
 *
 * 4 关于消费状态Consume Status
 * 对于并发的消费监听器，你可以返回 RECONSUME_LATER 来通知消费者现在不能消费这条消息，并且希望可以稍后重新消费它。然后，你可以继续消费其他消息。对于有序的消息监听器，因为你关心它的顺序，所以不能跳过消息，但是你可以返回SUSPEND_CURRENT_QUEUE_A_MOMENT 告诉消费者等待片刻。
 *
 * 5 关于Blocking
 * 不建议阻塞监听器，因为它会阻塞线程池，并最终可能会终止消费进程
 *
 * 6 关于线程数设置
 * 消费者使用 ThreadPoolExecutor 在内部对消息进行消费，所以你可以通过设置 setConsumeThreadMin 或 setConsumeThreadMax 来改变它。
 *
 * 7 关于消费位点
 * 当建立一个新的消费者组时，需要决定是否需要消费已经存在于 Broker 中的历史消息CONSUME_FROM_LAST_OFFSET 将会忽略历史消息，并消费之后生成的任何消息。CONSUME_FROM_FIRST_OFFSET 将会消费每个存在于 Broker 中的信息。你也可以使用 CONSUME_FROM_TIMESTAMP 来消费在指定时间戳后产生的消息。
 *
 *
 */
@Slf4j
public class SkipMessage {

    @Test
    public void producer() throws Exception {
        // 实例化消息生产者Producer
        DefaultMQProducer producer = new DefaultMQProducer("skip_producer_group_name");
        // 设置NameServer的地址
        producer.setNamesrvAddr("127.0.0.1:9876");
        // 启动Producer实例
        producer.start();
        new Thread(() -> {
            try {
                for (int i = 0; i < 1000; i++) {
                    // 创建消息，并指定Topic，Tag和消息体
                    Message msg = new Message(
                            "TopicTest" /* Topic */,
                            "TagA" /* Tag */,
                            ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
                    );

                    // 发送消息到一个Broker
                    SendResult sendResult = producer.send(msg);
                    // 通过sendResult返回消息是否成功送达
//                    log.info("{}", sendResult);
//                    TimeUnit.SECONDS.sleep(5);
                }
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }).start();

        // 如果不再发送消息，关闭Producer实例。
//        producer.shutdown();
    }

    @AfterEach
    public void consumer() throws Exception {
        // 实例化消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("skip_consumer_group_name");

        /**
         * 在此要说明一点，默认来说 消费者会获取 默认的4个队列中的消息，
         * 而消费者线程只存在一个的时候，是逐条queue去消费消息
         */
        consumer.setConsumeThreadMin(1);
        consumer.setConsumeThreadMax(1);
        // 设置NameServer的地址
        consumer.setNamesrvAddr("127.0.0.1:9876");

        // 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
        consumer.subscribe("TopicTest", "*");
        // 注册回调实现类来处理从broker拉取回来的消息
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            log.info("=======================================");
            log.info("Messages: {}", msgs);
            long offset = msgs.get(0).getQueueOffset();
            String maxOffset = msgs.get(0).getProperty(MessageConst.PROPERTY_MAX_OFFSET);
            log.info("offset : {} , maxOffset : {}", offset, maxOffset);
            long diff = Long.parseLong(maxOffset) - offset;
            if (diff > 5) {
                log.info("diff : {}", diff);
                // TODO 消息堆积情况的特殊处理
                msgs.stream().forEach(msg -> log.info("消息堆积情况的特殊处理 : {}", new String(msg.getBody())));
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            // TODO 正常消费过程
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        // 启动消费者实例
        consumer.start();
        log.info("Consumer Started.");

        TimeUnit.SECONDS.sleep(60);
    }

}
